package com.biz.primus.ms.search.service;

import com.biz.primus.base.constant.DepotStreamConstant;
import com.biz.primus.base.elasticsearch.entities.DepotEntity;
import com.biz.primus.base.elasticsearch.repository.DepotEsRepository;
import com.biz.primus.base.elasticsearch.util.PageUtil;
import com.biz.primus.base.vo.PageResult;
import com.biz.primus.base.vo.PageableRequestVo;
import com.biz.primus.base.vo.SearchResult;
import com.biz.primus.common.utils.DistanceCalc;
import com.biz.primus.common.utils.ValueUtils;
import com.biz.primus.model.depot.enums.BusinessStatus;
import com.biz.primus.model.depot.vo.*;
import com.biz.primus.ms.search.feign.DepotFeignClient;
import com.biz.primus.ms.search.transform.DepotEsIndexVo2DepotEntity;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.common.StopWatch;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.sort.GeoDistanceSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DepotSearchService {

    private final DepotEsRepository depotEsRepository;

    private final DepotFeignClient depotFeignClient;

    private DepotEsIndexVo2DepotEntity transformer = new DepotEsIndexVo2DepotEntity();

    @Autowired
    public DepotSearchService(DepotEsRepository depotEsRepository, DepotFeignClient depotFeignClient) {
        this.depotEsRepository = depotEsRepository;
        this.depotFeignClient = depotFeignClient;
    }


    public SearchResult<DepotSearchResultVo> searchDepot(SearchDepotConditionVo vo) {
        log.info("DepotEsRepositoryImpl#searchDepot vo: {}", vo.toString());
        if (vo.getLatitude() == null || vo.getLongitude() == null) {
            log.error("定位信息为空");
            return new SearchResult<>();
        }
        SearchResult<DepotSearchResultVo> searchResult = new SearchResult<>();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 过滤门店(只查询开业状态)
        TermQueryBuilder businessStatusTermQuery = QueryBuilders.termQuery("bizStatus", BusinessStatus.ON.getValue());
        boolQueryBuilder.must(businessStatusTermQuery);
        // 根据关键词模糊查询
        if (StringUtils.isNotBlank(vo.getKeyword())) {
            log.debug("DepotEsRepositoryImpl searchDepot keyword: {}", vo.getKeyword());
            QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery(vo.getKeyword());
            queryStringQueryBuilder.field("name").field("address").analyzer("ik_smart");
            boolQueryBuilder.must(queryStringQueryBuilder);
        }
        if (vo.getProvinceId() != null) {
            log.debug("查询门店限制省{}", vo.getProvinceId());
            TermQueryBuilder geoTermQuery = QueryBuilders.termQuery("provinceId", vo.getProvinceId());
            boolQueryBuilder.must(geoTermQuery);
        }
        // 根据距离过滤附近门店(只查询距离当前定位多少公里以内的门店)
        Double distance = vo.getDistance();
        Double lat = vo.getLatitude();
        Double lon = vo.getLongitude();
        if (ValueUtils.getValue(distance) > 0) {
            if (log.isDebugEnabled()) {
                log.debug("DepotEsRepositoryImpl searchDepot distance: {}", vo.getDistance());
            }
            boolQueryBuilder.must(QueryBuilders
                    .geoDistanceQuery("geoPoint")
                    .point(lat, lon)
                    .distance(distance, DistanceUnit.KILOMETERS));
        } else {
            log.debug("DepotEsRepositoryImpl searchDepot distance is null or 0");
        }
        Boolean pageable = vo.getPageable();
        GeoDistanceSortBuilder geoDistanceSortBuilder = SortBuilders.geoDistanceSort("geoPoint").point(lat, lon).order(SortOrder.ASC);
        SearchQuery searchQuery;
        if (pageable) {
            // 构造分页对象
            PageRequest pageRequest = PageUtil.getPageResult(vo.getPage(), vo.getPageSize());
            log.debug("DepotEsRepositoryImpl searchDepot pageRequest: {}", pageRequest.toString());
            searchQuery = new NativeSearchQueryBuilder()
                    .withQuery(boolQueryBuilder)
                    .withSort(geoDistanceSortBuilder)
                    .withPageable(pageRequest)
                    .build();
        } else {
            searchQuery = new NativeSearchQueryBuilder()
                    .withQuery(boolQueryBuilder)
                    .withSort(geoDistanceSortBuilder)
                    .build();
        }
        log.debug("DepotEsRepositoryImpl searchDepot searchQuery: {}", searchQuery.getQuery().toString());
        Page<DepotEntity> page = this.depotEsRepository.search(searchQuery);
        searchResult.setTotalNum(page.getTotalElements());
        if (CollectionUtils.isNotEmpty(page.getContent())) {
            List<DepotSearchResultVo> depotSearchResults = Lists.newArrayList();
            for (DepotEntity depotEntity : page.getContent()) {
                DepotSearchResultVo depotSearchResult = new DepotSearchResultVo();
                Double depotLat = depotEntity.getGeoPoint().getLat();
                Double depotLon = depotEntity.getGeoPoint().getLon();
                Double depotDistance = DistanceCalc.distance(lat, lon, depotLat, depotLon);
                depotSearchResult.setId(depotEntity.getId());
                depotSearchResult.setDistance(depotDistance.intValue());
                depotSearchResults.add(depotSearchResult);
            }
            searchResult.setItems(depotSearchResults);
        }
        log.debug("DepotEsRepositoryImpl searchDepot searchResult after search: {}", searchResult);
        return searchResult;
    }

    @StreamListener(DepotStreamConstant.DEPOT_CHANGE_ALL)
    public void updateTotalDepotIndexDocuments(DepotChangeAllMessageVo msgVo) {
        log.info("start updateTotalDepotIndexDocuments");
        StopWatch stopWatch = new StopWatch("更新全量门店索引");
        PageableRequestVo requestVo = new PageableRequestVo();
        Integer page = 0;
        Integer pageSize = 500;
        requestVo.setPage(page++);
        requestVo.setSize(pageSize);
        PageResult<DepotEsIndexVo> idxVo;
        long currentTimeMillis = System.currentTimeMillis();
        do {
            idxVo = depotFeignClient.createFullDepotEsVo(requestVo);
            List<DepotEsIndexVo> indexVos = idxVo.getContent();
            List<DepotEntity> entities = Lists.newArrayList();
            for (DepotEsIndexVo vo : indexVos) {
                if (vo != null) {
                    DepotEntity depotEntity = transformer.apply(vo);
                    assert depotEntity != null;
                    depotEntity.setLastUpdateTime(currentTimeMillis);
                    entities.add(depotEntity);
                }
            }
            depotEsRepository.save(entities);
            requestVo.setPage(page++);
        } while (idxVo.hasNext());
        StopWatch sw = new StopWatch("删除老版本索引数据");
        depotEsRepository.deleteByLastUpdateTimeLessThanEqual(currentTimeMillis - 1000);
        sw.stop();
        sw.prettyPrint();
        stopWatch.stop();
        stopWatch.prettyPrint();
    }

    @StreamListener(DepotStreamConstant.DEPOT_CHANGE)
    public void updateIncrDepotIndexDocument(DepotChangeMessageVo vo) {
        List<String> depotCodes = vo.getDepotCodes();
        Boolean deleted = vo.getDeleted();
        if (CollectionUtils.isEmpty(depotCodes)) {
            log.error("增量索引更新-->门店编码为空");
            return;
        }
        IncrementDepotEsIndexQueryVo incrVo = new IncrementDepotEsIndexQueryVo();
        incrVo.setDepotCodes(depotCodes);
        List<DepotEsIndexVo> idxVos = depotFeignClient.createIncrementDepotEsVo(incrVo);
        if (CollectionUtils.isEmpty(idxVos)) {
            log.error("depot 增量索引失败 {},没有找到门店", incrVo);
            return;
        }
        StopWatch stopWatch = new StopWatch("更新增量门店索引");
        if (deleted) {
            List<Long> ids = idxVos.stream().map(DepotEsIndexVo::getId).collect(Collectors.toList());
            ids.forEach(depotEsRepository::delete);
            log.debug("删除门店索引, ids: {}", ids.toString());
        } else {
            List<DepotEntity> entitys = idxVos.stream().map(new DepotEsIndexVo2DepotEntity()::apply).collect(Collectors.toList());
            depotEsRepository.save(entitys);
            log.debug("更新门店索引{}", entitys.size());
        }
        stopWatch.stop();
        stopWatch.prettyPrint();
    }

}
